package com.rickpan.service;

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

import com.rickpan.dto.response.TeamMessageDTO;
import com.rickpan.socketio.TeamEventHandler;
import com.rickpan.config.RabbitMQConfig;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 动态消费者管理器
 * 负责为用户动态创建和管理RabbitMQ消息队列的消费者
 * 
 * @author 开发团队协调员
 * @since 2025-07-19
 */
@Service
@ConditionalOnProperty(name = "spring.rabbitmq.host", matchIfMissing = false)
public class DynamicConsumerManager {

    private static final Logger logger = LoggerFactory.getLogger(DynamicConsumerManager.class);

    @Autowired
    private ConnectionFactory connectionFactory;
    
    @Autowired
    private RabbitAdmin rabbitAdmin;
    
    @Autowired
    private TeamEventHandler teamEventHandler;
    
    @Autowired
    private TeamMemberService teamMemberService;
    
    @Autowired
    private MessageQueueService messageQueueService;
    
    // 用户消费者容器映射 (userId -> container)
    private final Map<Long, SimpleMessageListenerContainer> userConsumerContainers = new ConcurrentHashMap<>();
    
    /**
     * 为用户创建队列消费者
     * 当用户进入聊天室时调用
     * 
     * @param userId 用户ID
     */
    public void createUserConsumer(Long userId) {
        if (userConsumerContainers.containsKey(userId)) {
            logger.debug("⚠️ 用户消费者已存在，跳过创建: userId={}", userId);
            return;
        }
        
        try {
            // 确保用户通知队列存在
            messageQueueService.ensureUserNotificationQueue(userId);
            
            // 创建消息监听器容器
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
            container.setConnectionFactory(connectionFactory);
            container.setQueueNames("user.notification." + userId);
            
            // 创建消息监听器适配器
            MessageListenerAdapter adapter = new MessageListenerAdapter(this, "handleUserMessage");
            // 设置消息转换器，确保正确反序列化TeamMessageDTO
            adapter.setMessageConverter(rabbitAdmin.getRabbitTemplate().getMessageConverter());
            container.setMessageListener(adapter);
            
            // 设置并发消费者数量
            container.setConcurrentConsumers(1);
            container.setMaxConcurrentConsumers(1);
            
            // 启动容器
            container.start();
            
            // 缓存容器
            userConsumerContainers.put(userId, container);
            
            logger.info("✅ 用户消费者创建成功: userId={}", userId);
            
        } catch (Exception e) {
            logger.error("❌ 用户消费者创建失败: userId={}", userId, e);
            throw new RuntimeException("用户消费者创建失败", e);
        }
    }
    
    /**
     * 销毁用户队列消费者
     * 当用户离开聊天室时调用
     * 
     * @param userId 用户ID
     */
    public void destroyUserConsumer(Long userId) {
        SimpleMessageListenerContainer container = userConsumerContainers.remove(userId);
        if (container != null) {
            try {
                container.stop();
                logger.info("✅ 用户消费者销毁成功: userId={}", userId);
            } catch (Exception e) {
                logger.error("❌ 用户消费者销毁失败: userId={}", userId, e);
            }
        } else {
            logger.debug("⚠️ 用户消费者不存在，跳过销毁: userId={}", userId);
        }
    }
    
    /**
     * 处理用户队列消息
     * 这个方法会被MessageListenerAdapter调用
     * 
     * @param message 消息内容
     */
    public void handleUserMessage(TeamMessageDTO message) {
        logger.info("📥 用户队列接收到团队消息: teamId={}, messageId={}", 
                   message.getTeamId(), message.getId());

        try {
            // 直接通过TeamEventHandler广播消息
            // 因为这是从用户队列消费的，所以直接广播给所有团队成员
            teamEventHandler.broadcastTeamMessageFromQueue(message);
            
            logger.info("✅ 用户队列消息处理成功: teamId={}, messageId={}", 
                       message.getTeamId(), message.getId());

        } catch (Exception e) {
            logger.error("❌ 用户队列消息处理失败: teamId={}, messageId={}", 
                        message.getTeamId(), message.getId(), e);
            // 这里可以添加重试逻辑或死信处理
        }
    }
    
    /**
     * 检查用户消费者是否存在
     * 
     * @param userId 用户ID
     * @return 是否存在
     */
    public boolean hasUserConsumer(Long userId) {
        return userConsumerContainers.containsKey(userId);
    }
    
    /**
     * 获取活跃的用户消费者数量
     * 
     * @return 消费者数量
     */
    public int getActiveConsumerCount() {
        return userConsumerContainers.size();
    }
    
    /**
     * 清理所有用户消费者
     * 应用关闭时调用
     */
    public void destroyAllConsumers() {
        logger.info("🧹 开始清理所有用户消费者，总数: {}", userConsumerContainers.size());
        
        userConsumerContainers.forEach((userId, container) -> {
            try {
                container.stop();
                logger.debug("✅ 用户消费者清理成功: userId={}", userId);
            } catch (Exception e) {
                logger.error("❌ 用户消费者清理失败: userId={}", userId, e);
            }
        });
        
        userConsumerContainers.clear();
        logger.info("✅ 所有用户消费者清理完成");
    }
}
